package dmp.beans.sparksql

import java.io.FileInputStream
import java.util.Properties

import org.apache.commons.configuration.ConfigurationFactory
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, SQLContext, SparkSession}

/*
广告投放地域统计
用sparksql实现
 */
object AreaAnalyseRpt_Disc {
  def main(args: Array[String]): Unit = {
//校验参数个数
    if(args.length!=2){
      println("参数设置不合法，退出程序")
      sys.exit()
    }
    val Array(inputPath,outputPath)=args
    val conf=new SparkConf().setMaster("local[*]").setAppName("AreaAnalyseRpt")
      .set("spark.serializer","org.apache.spark.serializer.KryoSerializer")//rdd序列化到磁盘，用于worker和worker之间的传输
    //"spark.serializer","org.apache.spark.serializer.KryoSerializer
    val spark=SparkSession.builder().config(conf).getOrCreate()
    //val sQLContext =new SQLContext(spark)

    //读取parquet文件
    val parquetData: DataFrame = spark.read.parquet(inputPath)
    //拿到数据后注册一张临时表
    parquetData.createOrReplaceTempView("log")
    val result: DataFrame = spark.sql(
      """
|select provincename,cityname,
|sum(case when requestmode = 1 and processnode >=1 then 1 else 0 end) as `原始请求`,
|sum(case when requestmode = 1 and processnode >=2 then 1 else 0 end) as `有效请求`,
|sum(case when requestmode = 1 and processnode = 3 then 1 else 0 end) as `广告请求`,
|sum(case when iseffective = 1 and isbilling = 1 and isbid = 1 then 1 else 0 end) as `参与竞价数`,
|sum(case when iseffective = 1 and isbilling = 1 and iswin = 1 and adorderid <> 0 then 1 else 0 end) as `竞价成功数`,
|sum(case when iseffective = 1 and isbilling = 1 and isbid = 1 then 1 else 0 end)/sum(case when iseffective = 1 and isbilling = 1 and iswin = 1 and adorderid <> 0 then 1 else 0 end) as `竞价成功率`,
|sum(case when requestmode = 2 and iseffective = 1 then 1 else 0 end) as `展示量`,
|sum(case when requestmode = 3 and iseffective = 1 then 1 else 0 end) as `点击量`,
|sum(case when requestmode = 2 and iseffective = 1 then 1 else 0 end)/sum(case when requestmode = 3 and iseffective = 1 then 1 else 0 end) as `点击率`,
|sum(case when iseffective = 1 and isbilling = 1 and iswin = 1 then winprice else 0 end)/1000 as `广告成本`,
|sum(case when iseffective = 1 and isbilling = 1 and iswin = 1 then adpayment else 0 end)/1000 as `广告消费`
|from log
|group by provincename,cityname order by provincename
      """.stripMargin
    )
    result.show()
    result.rdd.saveAsTextFile(outputPath)
    spark.stop()
  }
}
